package com.atguigu.dga.governance.assessor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.common.util.HttpUtil;
import com.atguigu.dga.ds.bean.TDsTaskInstance;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

import static com.atguigu.dga.common.constants.CommonConst.DS_TASK_STATE_SUCCESS;


@Component("IS_DATA_SKEW")
public class IsDataSkewAssessor    extends  Assessor{
    //1  取出任务的yarn_id

    //2  通过yarnid 访问日志服务 获得 attemptId

    //3  通过attemptId 访问 yarn 获得  stageId列表

    //4  循环stageId列表 获得各个stage任务列表

    //5  计算一个stage任务列表的duration  -》 最大duration 其他的平均值   最大值和其他平均的差值   差值的占比

    //6  评分 批语

    public static final String history_url_prefix="http://hadoop102:18080/api/v1/applications";
    @Override
    public void checkProblem(AssessParam assessParam, GovernanceAssessDetail governanceAssessDetail) throws Exception {
        //0 取参数
        String metricParamsJson = assessParam.getGovernanceMetric().getMetricParamsJson();
        JSONObject paramJsonObj = JSON.parseObject(metricParamsJson);

        //1  取出任务的yarn_id
        //当日最后一次成功的实例
        List<TDsTaskInstance> tDsTaskInstanceList = assessParam.getTDsTaskInstanceList();
        if(tDsTaskInstanceList==null) {
            return;
        }
        TDsTaskInstance lastTDsTaskInstance=null;
        for (TDsTaskInstance tDsTaskInstance : tDsTaskInstanceList) {
            if(tDsTaskInstance.getState().toString().equals(DS_TASK_STATE_SUCCESS)){
                if(lastTDsTaskInstance==null
                        ||tDsTaskInstance.getEndTime().getTime()>lastTDsTaskInstance.getEndTime().getTime()){
                    lastTDsTaskInstance=tDsTaskInstance;
                }
            }
        }
        String yarnId = lastTDsTaskInstance.getAppLink();

        //2  通过yarnid 访问日志服务 获得 attemptId
        String url = history_url_prefix+"/"+yarnId;
        String jobJson = HttpUtil.get(url);
        JSONObject jobJsonObj = JSON.parseObject(jobJson);
        JSONArray attemptJsonArray = jobJsonObj.getJSONArray("attempts");
        String lastAttempId = "0";
        for (int i = 0; i < attemptJsonArray.size(); i++) {
            JSONObject attemptJsonObj =  attemptJsonArray.getJSONObject(i);
            String attemptId = attemptJsonObj.getString("attemptId");
            boolean completed = attemptJsonObj.getBoolean("completed");
            if( attemptId.compareTo(lastAttempId) >0 && completed){
                lastAttempId = attemptId;
            }
        }
        if(lastAttempId.equals("0")){
            return;
        }
        //3  通过attemptId 访问 yarn 获得  stageId列表 (1 多task ,2 时长超过阈值的stage)
        String stagesUrl= history_url_prefix+"/"+yarnId+"/"+lastAttempId+"/stages";
        String stagesJson = HttpUtil.get(stagesUrl);
        List<JSONObject> stagesJSonObjList = JSON.parseArray(stagesJson, JSONObject.class);
//        List<Integer> stageIdList=new ArrayList<>();
//        for (JSONObject jsonObject : stagesJSonObjList) {
//
//        }

        List<Integer> stageIdList = stagesJSonObjList.stream().filter(jsonObj -> {
            //1 task数>=2保留 2  完成时间-提交时间>阈值保留
            String completionTime = jsonObj.getString("completionTime");
            String submissionTime = jsonObj.getString("submissionTime");
            if( completionTime==null||submissionTime==null){
                return false;
            }
            Long durationMs = 0L;
            try {
                Date completionDateTime = DateUtils.parseDate(completionTime, "yyyy-MM-dd'T'HH:mm:ss.SSS'GMT'");
                Date submissionDateTime = DateUtils.parseDate(submissionTime, "yyyy-MM-dd'T'HH:mm:ss.SSS'GMT'");
                durationMs = completionDateTime.getTime() - submissionDateTime.getTime();
            } catch (ParseException e) {
                throw new RuntimeException(e);
            }


            Long paramDurationMs = paramJsonObj.getLong("stage_dur_seconds");

            return jsonObj.getInteger("numTasks") >= 2 &&
                    durationMs * 1000 > paramDurationMs;

        }).map(jsonObj -> jsonObj.getInteger("stageId")).collect(Collectors.toList());


        //4  循环stageId列表 获得各个stage任务列表

        BigDecimal paramPercent = paramJsonObj.getBigDecimal("percent");
        List<Integer> dataSkewStageIdList = new ArrayList<>();
        for (Integer stageId : stageIdList) {
              String stageTaskUrl= history_url_prefix+"/"+yarnId+"/"+lastAttempId+"/stages/"+stageId;
            String stageTaskJson = HttpUtil.get(stageTaskUrl);
            List<JSONObject> stageTaskJsonObjList = JSON.parseArray(stageTaskJson, JSONObject.class);
            JSONObject stageTaskJsonObj = stageTaskJsonObjList.get(0);
            JSONObject tasksJsonObj = stageTaskJsonObj.getJSONObject("tasks");
            List<Integer> taskDurationList= new ArrayList<>(); //存储每个task的duration
            for (Object taskObj : tasksJsonObj.values()) {
               JSONObject taskJsonObj=    (JSONObject) taskObj;
                Integer durationMs = taskJsonObj.getInteger("duration");
                taskDurationList.add(durationMs);
            }
            //判断是否有数据倾斜
            boolean isDataSkew = checkDataSkew(taskDurationList, paramPercent);
            if(isDataSkew){
               dataSkewStageIdList.add(stageId);
            }

        }

        //6  评分 批语
        if(dataSkewStageIdList.size()>0){
            governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
            governanceAssessDetail.setAssessProblem("存在数据倾斜：stageIds:"+ StringUtils.join(dataSkewStageIdList,","));
        }



    }

    //如果某个stage的最大任务耗时超过其他任务平均耗时耗时的{percent}%
    private boolean  checkDataSkew(List<Integer> taskDurationList, BigDecimal paramPercent){
        Integer maxDuration=0;
        Integer sumDuration=0;
        for (Integer duration : taskDurationList) {
            if(duration>maxDuration) {
                maxDuration=duration;
            }
            sumDuration+=duration;
        }
        Integer avgOtherDuration=(sumDuration-maxDuration)/(taskDurationList.size()-1);
        //求超过的百分比
        Integer overDurationMs = maxDuration - avgOtherDuration;
        BigDecimal overPercent = BigDecimal.valueOf(overDurationMs).movePointRight(2).divide(BigDecimal.valueOf(avgOtherDuration) , 1, BigDecimal.ROUND_HALF_UP)  ;

        return overPercent.compareTo(paramPercent)>0 ;

    }
}
